package core

import (
	"encoding/json"
	"fmt"
	"gark-elastic-job/common"
	logger "github.com/sirupsen/logrus"
)

// JobExecutor就是调度的执行器, 将每个任务放到合适的etcd目录下
type JobExecutor struct {
	node      *JobNode
	snapshots chan *common.GarkJobSnapshot
}

func NewExecutor(node *JobNode) *JobExecutor {
	logger.Debug("start Job Executor")
	jobExecutor := &JobExecutor{
		node:      node,
		snapshots: make(chan *common.GarkJobSnapshot, 512),
	}
	go jobExecutor.exec()
	return jobExecutor
}

func (jobExecutor *JobExecutor) exec() {
	for jobSnapshot := range jobExecutor.snapshots {
		jobExecutor.handleJobSnapshot(jobSnapshot)
	}
}

// handleJobSnapshot to put an executable job to appropriate position(client)
func (jobExecutor *JobExecutor) handleJobSnapshot(snapshot *common.GarkJobSnapshot) {
	// 选择一个合适的客户端
	group := snapshot.Group
	client, err := jobExecutor.node.GroupManager.selectClient(group)
	if err != nil {
		logger.Errorf("group manager failed to select client, error: %v", err)
		return
	}

	clientId := client.ID
	snapshot.IP = client.IP

	value, err := json.Marshal(snapshot)
	if err != nil {
		logger.Warnf("encode snapshot: %v failed, error: %v", snapshot, err)
		return
	}

	snapshotPath := fmt.Sprintf(JobClientSnapshotPath, group, clientId, snapshot.Id)
	if err = jobExecutor.node.etcd.Put(snapshotPath, string(value)); err != nil {
		logger.Errorf("failed to put snapshot: %v to client:%v, error: %v", snapshot, client, err)
		return
	}

}

func (jobExecutor *JobExecutor) pushJobSnapshot(snapshot *common.GarkJobSnapshot) {
	jobExecutor.snapshots <- snapshot
}
